import os
import pyspark.sql.functions
from pyspark.sql import SparkSession, DataFrame
from pyspark.sql.types import StringType

from tag.ESMeta import ruleToESMeta

# 0.设置系统环境变量
os.environ['JAVA_HOME'] = '/export/server/jdk1.8.0_241/'
os.environ['SPARK_HOME'] = '/export/server/spark'
os.environ['PYSPARK_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'
os.environ['PYSPARK_DRIVER_PYTHON'] = '/root/anaconda3/envs/pyspark_env/bin/python'

@pyspark.sql.functions.udf(returnType=StringType())
def merge_tags(new_tags, old_tags):
    #1.如果old_tags为空，则返回new_tags
    if old_tags is None:
        return new_tags
    #2.如果new_tags为空，则返回old_tags
    elif new_tags is None:
        return old_tags
    #3.如果都不为空，则进行合并
    else:
        #对new_tags进行切分
        new_tags_list = str(new_tags).split(",")
        #对old_tags进行切分
        old_tags_list = str(old_tags).split(",")
        #对切分后的结果进行合并
        result_tags = new_tags_list + old_tags_list
        #把列表拼接成字符串返回
        return ','.join(set(result_tags))

class BaseModel(object):
    """
        用户自定义的基类，在这个类中定义标签的步骤
        第一部分：
            #0.初始化Spark环境
            #1.读取MySQL标签体系数据
            #2.过滤4级标签的数据，将四级标签的rule转换为esMeta对象
            #3.根据esMeta对象从ES中读取相应的业务数据
            #4.根据4级标签ID，读取5级标签的数据
            #5.通过ES中的业务数据与MySQL的5级标签进行打标签
            #6.从ES中读取历史用户标签数据
            #7.将老的用户画像标签与新的标签进行合并，得到最终标签
            #8.将最终的结果写入ES中
            #9.销毁Spark环境，释放资源
        #第二部分
            #额外定义一个函数（方法），把上述10个函数串联起来执行
            #1.读取MySQL标签体系数据
            #2.过滤4级标签的数据，将四级标签的rule转换为esMeta对象
            #3.根据esMeta对象从ES中读取相应的业务数据
            #4.根据4级标签ID，读取5级标签的数据
            #5.通过ES中的业务数据与MySQL的5级标签进行打标签
            #6.从ES中读取历史用户标签数据
            #7.将老的用户画像标签与新的标签进行合并，得到最终标签
            #8.将最终的结果写入ES中
            #9.销毁Spark环境，释放资源
        """
    #  0.初始化Spark环境
    def __init__(self,fourId):
        "把spark变量声明为类的属性"
        self.spark=SparkSession.builder.master("local[2]").appName("SparkSQLAppName").config("spark.sql.shuffle.partitions",4).getOrCreate()

        self.fourId=fourId

    #  1.读取MySQL标签体系数据
    def read_from_mysql(self):
        #连接MySQL
        input_df=self.spark.read.jdbc(
            url='jdbc:mysql://up01:3306/tfec_tags',
            table='tbl_basic_tag',
            properties={'user':'root','password':'123456'}
        )
        return input_df
    #  2.过滤4级标签的数据，将四级标签的rule转换为esMeta对象
    def rule_to_es_meta(self,input_df):
        four_df=input_df.where(f'id={self.fourId}').select('rule')
        # 获取rule字符串数据
        rule=four_df.first()['rule']
        # 把rule字符串转换成ESMeta
        esMeta=ruleToESMeta(rule)
        return esMeta
    #  3.根据esMeta对象从ES中读取相应的业务数据
    def read_from_es(self, esMeta):
        es_df = self.spark.read.format("es") \
            .option("es.resource", esMeta.esIndex) \
            .option("es.nodes", esMeta.esNodes) \
            .option("es.read.field.include", esMeta.selectFields) \
            .option("es.statistics.date.rich", "False") \
            .load()
        return es_df
    #  4.根据4级标签ID，读取5级标签的数据
    def get_five_tags(self, input_df):
        five_df:DataFrame= input_df.where(f"pid = {self.fourId}").select("id", "rule")

        return five_df
    #  5.通过ES中的业务数据与MySQL的5级标签进行打标签
        # 打标签需要业务数据（es_df）和标签规则数据（five_df）
        # 当这个方法在子类实现后，也就是标签打完了，要记得返回新的标签结果（new_df）
    def compute(self, es_df, five_df):
        pass
    #  6.从ES中读取历史用户标签数据
    def read_old_df_from_es(self, esMeta):
        old_df = self.spark.read \
            .format("es") \
            .option("es.resource", "tags_result") \
            .option("es.nodes", esMeta.esNodes) \
            .load()
        return old_df
    #  7.将老的用户画像标签与新的标签进行合并，得到最终标签
    def merge_new_df_and_old_df(self, new_df, old_df):
        result_df = new_df.join(other=old_df,
                                on=new_df['userId'] == old_df['userId'],
                                how='left') \
            .select(new_df['userId'].cast(StringType()),
                    merge_tags(new_df['tagsId'], old_df['tagsId']).alias("tagsId"))
        return result_df
    #  8.将最终的结果写入ES中
    def save_result_df_to_es(self, result_df, esMeta):
        result_df.write \
            .mode("append") \
            .format("es") \
            .option("es.resource", "tags_result") \
            .option("es.nodes", esMeta.esNodes) \
            .option("es.statistics.id", "userId") \
            .save()
    #  9.销毁Spark环境，释放资源
    def close(self):
        self.spark.stop()

    # TODO 把以上9个方法串联起来一起运行
    # 额外定义一个函数（方法），把上述10个函数串联起来执行
    def execute(self):
        # 1.读取MySQL标签体系数据
        input_df=self.read_from_mysql()
        # 2.过滤4级标签的数据，将四级标签的rule转换为esMeta对象
        esMeta=self.rule_to_es_meta(input_df)
        # 3.根据esMeta对象从ES中读取相应的业务数据
        es_df=self.read_from_es(esMeta)
        # 4.根据4级标签ID，读取5级标签的数据
        five_df=self.get_five_tags(input_df)
        # 5.通过ES中的业务数据与MySQL的5级标签进行打标签
        # 当这个方法在子类实现后，也就是标签打完了，要记得返回新的标签结果（new_df）
        new_df=self.compute(es_df,five_df)
        try:
            # 6.从ES中读取历史用户标签数据
            old_df=self.read_old_df_from_es(esMeta=esMeta)
            # 7.将老的用户画像标签与新的标签进行合并，得到最终标签
            result_df=self.merge_new_df_and_old_df(new_df, old_df)
        except:
            print("-------------首次运行，跳过合并-------------")
            result_df=new_df
        # 8.将最终的结果写入ES中
        self.save_result_df_to_es(result_df, esMeta)
        # 9.销毁Spark环境，释放资源
        self.close()


























